package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class Demo8WIndowApply {
    public static void main(String[] args) throws Exception {

        /*
         * 117988031603010,1614711895,36.38
         * 117988031603010,1614711904,35.38
         * 117988031603010,1614711914,45.38
         * 117988031603010,1614711924,45.29
         * 117988031603010,1614712022,75.29
         * 117988031603010,1614712120,46.29
         * 117988031603010,1614712218,82.29
         * 117988031603010,1614712316,82.29
         * 117988031603010,1614712414,48.5
         * 117988031603010,1614712619,59.5
         * 117988031603010,1614712824,52.5
         * 117988031603010,1614713030,71.5
         * 117988031603010,1614713235,65.5
         * 117988031603010,1614713441,78.26
         * 117988031603010,1614713547,76.26
         * 117988031603010,1614713653,108.26
         * 117988031603010,1614813653,108.26
         */

        /* 判断卡口是否拥堵
         * 每隔1分钟统计最近15分钟卡口的平均车速
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读取数据
        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //解析数据
        DataStream<Cars> carsDS = linesDS.map(line -> {
            String[] split = line.split(",");
            String card = split[0];
            long ts = Long.parseLong(split[1]) * 1000;
            double speed = Double.parseDouble(split[2]);
            return new Cars(card, ts, speed);
        });

        //指定时间字段和水位线
        DataStream<Cars> assDS = carsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Cars>forMonotonousTimestamps()
                        //指定时间字段
                        .withTimestampAssigner((car, ts) -> car.getTs())
        );

        //按照卡口分钟
        KeyedStream<Cars, String> keyByDS = assDS.keyBy(Cars::getCard);

        //划分窗口
        //每隔1分钟统计最近15分钟卡口的平均车速
        WindowedStream<Cars, String, TimeWindow> windowDS = keyByDS.window(SlidingEventTimeWindows.of(Time.minutes(15), Time.minutes(1)));

        /*
         * apply: 再划分窗口之后，再窗口内进行复杂计算您的算子
         */
        DataStream<Tuple2<String, Double>> avgSpeedDS = windowDS
                .apply(new WindowFunction<Cars, Tuple2<String, Double>, String, TimeWindow>() {
                    /**
                     * apply方法，每个key每个窗口执行一次
                     * @param card: 分组的key
                     * @param window : 窗口对象
                     * @param cars： 窗口内所有的数据
                     * @param out ：用于将数据发送到下游
                     */
                    @Override
                    public void apply(String card, TimeWindow window, Iterable<Cars> cars, Collector<Tuple2<String, Double>> out) throws Exception {
                        double sumSpeed = 0.0;
                        long flow = 0;
                        for (Cars car : cars) {
                            sumSpeed += car.getSpeed();
                            flow++;
                        }
                        //平均车速
                        double avgSpeed = sumSpeed / flow;

                        //将数据发送到下游
                        out.collect(Tuple2.of(card, avgSpeed));
                    }
                });

        avgSpeedDS.print();

        env.execute();
    }
}

class Cars {
    private String card;
    private long ts;
    private double speed;

    @Override
    public String toString() {
        return "Cars{" +
                "card='" + card + '\'' +
                ", ts=" + ts +
                ", speed=" + speed +
                '}';
    }

    public Cars() {
    }

    public Cars(String card, long ts, double speed) {
        this.card = card;
        this.ts = ts;
        this.speed = speed;
    }

    public String getCard() {
        return card;
    }

    public void setCard(String card) {
        this.card = card;
    }

    public long getTs() {
        return ts;
    }

    public void setTs(long ts) {
        this.ts = ts;
    }

    public double getSpeed() {
        return speed;
    }

    public void setSpeed(double speed) {
        this.speed = speed;
    }
}